package org
import org.apache.spark.sql.SparkSession

import java.util.regex.Pattern
object SparkYun2_Movies {

 def main(args: Array[String]): Unit = {
   //创建spark运行环境
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("spark")
      .getOrCreate()
    val sc = spark.sparkContext
    //电影用户分析数据概况
    val filePath="src/main/resources/"
    val moviesRDD =sc.textFile(filePath + "movies.dat")
    val occupationsRDD =sc.textFile(filePath + "occupations.dat")
    val ratingsRDD = sc.textFile(filePath + "ratings.dat")
    val usersRDD = sc.textFile(filePath + "users.dat")
    println("总电影数:"+moviesRDD.count())
    println("总用户数:"+usersRDD.count())
    println("评分条数:"+ ratingsRDD.count())
    println("职业数:"+ occupationsRDD.count())
    //用户分析：统计某用户评过分的电影数量并打印电影详情
    val userRating = ratingsRDD.map(_.split("::"))
      .map(user =>(user(1),user(0)))//转换为KEY（电影ID)value(电影ID）
      .filter(_._2.equals("2688")) //统计用户ID为2688的影评数量
  println("用户ID为2688的影评数量为:"+ userRating.count())
  val movieInfo = moviesRDD.map(_.split("::"))
    .map(movie =>(movie(0),(movie(1),movie(2))))//将电影数据转换为key value
    .join(userRating)
    .map(item => (item._1,item._2._2,item._2._1._1,item._2._1._2))
  movieInfo.take(10).foreach(println)


  val rating = ratingsRDD.map(x => x.split("::")).map {
   x => {
    (x(0), x(1), x(2)) // (UserID,MovieID,Rating)
      }
     }.cache()
    println("平均得分最高的前 10 名的电影名称简单版")
  rating.map(x => (x._2, (x._3.toDouble, 1)))
  //  ( MovieId,( 总分, 总次数) )
    .reduceByKey((x, y) => {
   (x._1 + y._1, x._2 + y._2)
  }) // (  平均分, MovieId )
   .map(x => (x._2._1 / x._2._2, x._1))
  .sortByKey(false)
   .take(5)
   .foreach(println)
  println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次 数))")
  //MovieID::Title::Genres
  val moviesInfo = moviesRDD.map(x => x.split("::"))
   .map(x => {
     (x(0), (x(1), x(2)))
   })

  val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
  // ( MovieId,( Rating, 1) ) (1252,(4.0,1))
  .reduceByKey((x, y) => {
   (x._1 + y._1, x._2 + y._2)  }) // ( MovieId,( 总分, 总次数) )
   .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2))) //  ( MovieId, (平均分, 总分,总次数) )
  moviesInfo.join(ratingsInfo)
  .map(info => {
   (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
    // (平均分,(movieId,Title,Genres,总分,总次数))
   })
    .sortByKey(false)
    .take(5)
    .foreach(println)
   println("观影人数最多的前 10 部电影")
  // (UserID,MovieID,Rating)
  val watchViewsInfo = rating.map(x => {
  (x._2, 1)

 }).reduceByKey((x, y) => x + y) //  ( MovieId,总次数 )
   .map(x => (x._2, x._1))
    .sortByKey(false)
    .take(5) // 5 名
  watchViewsInfo.foreach(println(_))
  println("===================>")
 rating.map(x => (x._2, 1)) // ( MovieId, 1)
   .reduceByKey((x, y) => {
    (x + y)
   })// ( MovieId,总次数 )
  // .map(x => (x._2, x._1)) //   (  总次数, MovieId )
  .sortByKey(false)
   .take(10)
   .foreach(println) //  286-> 999
 println("详情的输出(  观影人数，电影编号)")
 // 输出 ( 总次数, (MovieID, title,Genres,总分,平均分  ) )
  //  ratingsInfo  ( MovieId, (平均分, 总分,总次数) )
  //  ( MovieID,  ( (Title,Genres),(平均分, 总分,总次数)  ) )
  // ratingsInfo ( MovieId, (平均分, 总分,总次数) )
  // moviesInfo  MovieID::Title::Genres
  moviesInfo.join(ratingsInfo).map(x => {
   (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))
  }).sortByKey(false)
    .take(10)
    .foreach(println)


val movieStyle = moviesRDD.map(_.split("::"))
  .map(x => (x(0),x(2)))//只留ID和电影数量类型
  .flatMapValues(x => x.split("\\|"))//根据符号拆分
  .map(x => (x._2,1))//转换为key value类型
  .reduceByKey((x,y)=>x+y)
  //在控制台打印科幻片的电影数量
  movieStyle.take(10).foreach(println)
 // movieStyle.foreach(println)
  //niandu电影数量统计
  //正则表达式 找出小括号里的四位数
 /* val pattern = Pattern.compile("(.*) (\\(\\d{4}\\))")

   val movieYear = moviesRDD.map(_.split("::"))
     .map(x=>(x(1),1))
     .map(x=>{
      var name=""
      var year=""
      val matcher = pattern matcher(x._1)
      if (matcher.find()){
       name=matcher.group(1)
       year=matcher.group(2)
       year=year.substring(1,year.length-1)
      }
      if (year ==""){
       (-1,1)
      }else{
        (year.toInt)

     }
     })
     .reduceByKey((x,y)=>x+y)
    movieYear.foreach(println)
    sc.stop()*/
  }

}
